#pragma once 
#include <iostream>
#include <vector>
#include <unordered_map>
#include <functional>
#include <memory>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <typeinfo>
#include <cassert>
#include <ctime>
#include <cerrno>
#include <cstring>
#include <cstdlib>
#include <unistd.h>
#include <signal.h>
#include <fcntl.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>


/******************************** 
            日志宏 
********************************/
#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL INF
#define LOG(level, format, ...) do {\
        if (level >= LOG_LEVEL) {\
            time_t t = time(NULL);\
            struct tm *ltm = localtime(&t);\
            char tmp[32] = {0};\
            strftime(tmp, 31, "%H:%M:%S", ltm);\
            fprintf(stdout, "[%p %s %s:%d]" format "\n", (void *)pthread_self(), tmp, __FILE__, __LINE__, ##__VA_ARGS__);\
        }\  
    }while(0)
#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)


/******************************** 
            Buffer 
********************************/
const int BUFFER_DEFAULT_SIZE = 1024;
class Buffer
{
private:
    std::vector<char> m_buffer; // 缓冲区  
    uint64_t m_reader_idx;      // 读偏移
    uint64_t m_writer_idx;      // 写偏移
private:
    char *Begin() { return &(*m_buffer.begin()); }
public:
    Buffer() :m_buffer(BUFFER_DEFAULT_SIZE), m_reader_idx(0), m_writer_idx(0){}
    // 获取当前写入地址
    char *WriterPosition() { return Begin() + m_writer_idx; }
    // 获取当前读取地址
    char *ReaderPosition() { return Begin() + m_reader_idx; }
    // 获取缓冲区末尾空闲空间大小
    uint64_t TailIdleSize() { return m_buffer.size() - m_writer_idx; }
    // 获取缓冲区起始空闲空间大小
    uint64_t HeadIdlesize() { return m_reader_idx; }
    // 获取可读数据大小
    uint64_t ReadAbleSize() { return m_writer_idx - m_reader_idx; }
    // 将读数据向后偏移
    void MoveReadOffset(uint64_t len)
    {
        if (len == 0) return;
        assert(len <= ReadAbleSize());
        m_reader_idx += len; 
    }
    // 将写数据向后偏移
    void MoveWriteOffset(uint64_t len)
    {
        // 向后移动的距离必须小于当前后面的空闲空间的大小
        assert(len <= TailIdleSize());
        m_writer_idx += len;
    }
    // 确保可写空间足够
    void EnsureWriteSpace(uint64_t len)
    {
        // 如果写的长度小于缓冲区尾部空闲时间，够
        if (len <= TailIdleSize()) return;
        // 末尾空间大小不够，加上开头空闲空间，如果够，将数据移到开头
        if (len <= TailIdleSize() + HeadIdlesize())
        {
            uint64_t rsz = ReadAbleSize();
            std::copy(ReaderPosition(), ReaderPosition() + rsz, Begin());
            m_reader_idx = 0;
            m_writer_idx = rsz;
        }
        else // 如果末尾空间大小加上开头空间空间不够，直接在后面开辟空间 
        {
            m_buffer.resize(m_writer_idx + len);
        }
    }
    // 写入数据
    void Write(const void *data, uint64_t len)
    {
        if (len == 0) return;
        EnsureWriteSpace(len);
        const char *d = (const char *)data;
        std::copy(d, d + len, WriterPosition());
    }
    void WriteAndPush(const void *data, uint64_t len)
    {
        Write(data, len);
        MoveWriteOffset(len);
    }
    void WriteString(const std::string& data) { Write(data.c_str(), data.size()); }
    void WriteStringAndPush(const std::string& data)
    {
        WriteString(data);
        MoveWriteOffset(data.size());
    }
    void WriteBuffer(Buffer& data) { Write(data.ReaderPosition(), data.ReadAbleSize()); }
    void WriteBufferAndPush(Buffer& data)
    {
        WriteBuffer(data);
        MoveWriteOffset(data.ReadAbleSize()); /* 注意！！！ 可能有误*/
    }
    // 读取数据
    void Read(void *buf, uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::copy(ReaderPosition(), ReaderPosition() + len, (char *)buf);
    }
    void ReadAndPop(void *buf, uint64_t len)
    {
        Read(buf, len);
        MoveReadOffset(len);
    }
    std::string ReadAsString(uint64_t len)
    {
        std::string str;
        str.resize(len);
        // 不能用str.c_str()，因为这是const char *类型
        Read(&str[0], len);
        return str;
    }
    std::string ReadAsStringAndPop(uint64_t len)
    {
        std::string str = ReadAsString(len);
        MoveReadOffset(len);
        return str;
    }
    // 清空缓冲区
    void Clear()
    {
        m_reader_idx = 0;
        m_writer_idx = 0;
    }
    // 读取一行数据
    char *FindCRLF()
    {
        char *res = (char *)memchr(ReaderPosition(), '\n', ReadAbleSize());
        return res;
    }
    std::string GetLine()
    {
        char *pos = FindCRLF();
        if (pos == nullptr)
            return "";
        return ReadAsString(pos - ReaderPosition() + 1);
    }
    std::string GetLineAndPop()
    {
        std::string str = GetLine();
        MoveReadOffset(str.size());
        return str;
    }
};


/******************************** 
            Socket 
********************************/
const int MAX_LISTEN = 1024;

class Socket
{
private:
    int m_sockfd;
public:
    Socket() : m_sockfd(-1) {}
    Socket(int fd) :m_sockfd(fd) {}
    ~Socket() { Close(); }
    int Fd() { return m_sockfd; }
    // 创建套接字
    bool Create()
    {
        // int socket(int domain, int type, int protocol);
        m_sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (m_sockfd < 0)
        {
            ERR_LOG("Create Socket Failed!");
            return false;
        }
        return true;
    }
    // 绑定地址信息
    bool Bind(const std::string& ip, uint16_t port)
    {
        sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        //inet_pton(AF_INET, ip.c_str(), &addr.sin_addr);
        socklen_t len = sizeof(addr);
        //int bind(int sockfd, const struct sockaddr *addr, socklen_t addrlen);
        int ret = bind(m_sockfd, (struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            ERR_LOG("Bind Socket Failed!");
            return false;
        }
        return true;
    }
    // 开始监听
    bool Listen(int backlog = MAX_LISTEN)
    {
        // int listen(int sockfd, int backlog);
        int ret = listen(m_sockfd, backlog);
        if (ret < 0)
        {
            ERR_LOG("Listen Socket Failed!");
            return false;
        }
        return true;
    }
    // 向服务器发起连接
    bool Connect(const std::string& ip, uint16_t port)
    {
        sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        inet_pton(AF_INET, ip.c_str(), &addr.sin_addr);
        socklen_t len = sizeof(addr);
        // int connect(int sockfd, const struct sockaddr *addr, socklen_t addrlen);
        int ret = connect(m_sockfd, (struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            ERR_LOG("Connect Sever Failed!");
            return false;
        }
        return true;
    }
    // 获取新连接
    int Accept()
    {
        //  int accept(int sockfd, struct sockaddr *addr, socklen_t *addrlen);
        int newfd = accept(m_sockfd, nullptr, nullptr);
        if (newfd < 0)
        {
            ERR_LOG("Accept Client Failed!");
            return -1;
        }
        return newfd;
    }
    // 接收数据
    ssize_t Recv(void *buf, size_t len, int flag = 0)
    {
        // ssize_t recv(int sockfd, void *buf, size_t len, int flags);
        ssize_t n = recv(m_sockfd, buf, len, flag);
        if (n <= 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            ERR_LOG("Recv Failed!");
            return -1;
        }
        return n;
    }
    // 非阻塞读取
    ssize_t NonBlockRecv(void *buf, size_t len)
    {
        return Recv(buf, len, MSG_DONTWAIT);
    }
    // 发送数据
    ssize_t Send(const void *buf, size_t len, int flag = 0)
    {
        // ssize_t send(int sockfd, const void *buf, size_t len, int flags);
        ssize_t n = send(m_sockfd, buf, len, flag);
        if (n < 0)
        {
            ERR_LOG("Send Failed!");
            return -1;
        }
        return n;
    }
    // 非阻塞发送
    ssize_t NonBlockSend(const void *buf, size_t len, int flag = 0)
    {
        if (len == 0)
            return 0;
        return Send(buf, len, MSG_DONTWAIT);
    }
    // 关闭套接字
    void Close()
    {
        if (m_sockfd != -1)
        {
            close(m_sockfd);
            m_sockfd = -1;
        }
    }
    // 创建一个监听连接（集成功能）
    bool CreateServer(uint16_t port, const std::string& ip = "0.0.0.0", bool block_flag = false)
    {
        // 创建、绑定、监听、设置非阻塞，地址重用
        if (!Create()) return false;
        if (block_flag) NonBlock();
        ReuseAddress();
        if (!Bind(ip, port)) return false;
        if (!Listen()) return false;
        return true;
    }
    // 创建一个客户端连接（集成功能）
    bool CreateClient(uint16_t port, const std::string& ip)
    {
        if (!Create()) return false;
        if (!Connect(ip, port)) return false;
        return true; 
    }
    // 设置套接字选项 --- 开启地址重用
    void ReuseAddress()
    {
        // int setsockopt(int sockfd, int level, int optname, const void *optval, socklen_t optlen);
        // 地址重用
        int val = 1;
        setsockopt(m_sockfd, SOL_SOCKET, SO_REUSEADDR, (void *)&val, sizeof(val));
        // 端口重用
        val = 1;
        setsockopt(m_sockfd, SOL_SOCKET, SO_REUSEPORT, (void *)&val, sizeof(val));
    }
    // 设置套接字阻塞属性 --- 设置为非阻塞
    void NonBlock()
    {
        //  int fcntl(int fd, int cmd, ... /* arg */ );
        int flag = fcntl(m_sockfd, F_GETFL, 0);
        fcntl(m_sockfd, F_SETFL, flag | O_NONBLOCK);
    }
};


/******************************** 
    Channel: 管理描述符 
********************************/
class Poller;
class EventLoop;
class Channel
{
private:
    int m_fd;           // 管理的描述符
    uint32_t m_events;  // 当前需要监控的事件 
    uint32_t m_revents; // 当前触发的事件
    EventLoop *m_loop;
    using EventCallBack = std::function<void()>;
    EventCallBack m_read_callback;   // 可读事件触发的回调函数
    EventCallBack m_write_callback;  // 可写事件触发的回调函数
    EventCallBack m_error_callback;  // 错误时间触发的回调函数
    EventCallBack m_close_callback;  // 连接断开触发的回调函数
    EventCallBack m_event_callback;  // 任意事件触发的回调函数

public:
    Channel(EventLoop *loop, int fd) :m_loop(loop), m_fd(fd), m_events(0), m_revents(0) {}
    ~Channel() {}

    int Fd() { return m_fd; }
    uint32_t Event() { return m_events; }
    // 记录触发事件
    void SetREvents(uint32_t events) { m_revents = events; }
    // 设置回调函数
    void SetReadCallBack(const EventCallBack& cb) { m_read_callback = cb; }
    void SetWriteCallBack(const EventCallBack& cb) { m_write_callback = cb; }
    void SetErrorCallBack(const EventCallBack& cb) { m_error_callback = cb; }
    void SetCloseCallBack(const EventCallBack& cb) { m_close_callback = cb; }
    void SetEventCallBack(const EventCallBack& cb) { m_event_callback = cb; }

    bool ReadAble() { return (m_events & EPOLLIN); }       // 当前事件是否可读
    bool WriteAble() { return (m_events & EPOLLOUT); }      // 当前事件是否可写
    void EnableRead() { m_events |= EPOLLIN; Update(); }             // 启动读事件关心
    void EnableWrite() { m_events |= EPOLLOUT; Update(); }           // 启动写事件关心
    void DisableRead() { m_events &= (~EPOLLIN); Update(); }         // 取消读事件关心
    void DisableWrite() { m_events &= (~EPOLLOUT); Update(); }       // 取消写事件关心
    void DisableAll() { m_events = 0; }                    // 取消所有事件关心
    void Remove();        // 移除监控 --- 也就是从epoll红黑树中删除某个节点
    void Update();
    void HandleEvent()    // 事件处理 --- 一旦触发了事件，调用这个函数，根据不同的事件，调用不同函数
    {
        if ((m_revents & EPOLLIN) || (m_revents & EPOLLPRI) || (m_revents & EPOLLRDHUP))
        {
            if (m_read_callback)
                m_read_callback();
        }
        /* 有可能会释放连接的事件，一次只处理一个 */
        if (m_revents & EPOLLOUT)
        {
            if (m_write_callback)
                m_write_callback();
        }
        else if (m_revents & EPOLLERR)
        {
            /* 一旦出错就会释放连接，所以任何事件的回调只能放在前面 */
            if (m_error_callback)
                m_error_callback();
        }
        else if (m_revents & EPOLLHUP)
        {
            /* 一旦出错就会释放连接，所以任何事件的回调只能放在前面 */
            if (m_close_callback)
                m_close_callback();
        }
        /* 不管任何事件，都会调用的回调函数 */
        if (m_event_callback)
            m_event_callback();
    }
};


/******************************** 
    Poller: 对描述符IO监控的管理
********************************/
const int MAX_POLLEREVENTS = 1024;
class Poller
{
private:
    int m_epfd;
    struct epoll_event m_evs[MAX_POLLEREVENTS];
    std::unordered_map<int, Channel *> m_channels;
private:
    // 对epoll直接操作
    void Update(Channel *channel, int op)
    {
        // int epoll_ctl(int epfd, int op, int fd, struct epoll_event *event);
        struct epoll_event event;
        event.data.fd = channel->Fd();
        event.events = channel->Event();
        int ret = epoll_ctl(m_epfd, op, channel->Fd(), &event);
        if (ret < 0)
        {
            ERR_LOG("EPOLLCTL FAILED!");
        }
        return;
    }
    // 判断一个Channel是否已经添加了事件监控
    bool HasChannel(Channel * channel)
    {
        
        auto it = m_channels.find(channel->Fd() m_channels.end());
        if (it != m_channels.end())
        {
            return true;
        }
        return false;   
    }
public:
    Poller() 
    {
        // 创建epoll
        m_epfd = epoll_create(MAX_POLLEREVENTS);
        if (m_epfd < 0)
        {
            ERR_LOG("EPOLL CREATE FAILED!");
            abort(); // 程序退出
        }
    }
    ~Poller() {}
    // 添加/修改监控事件
    void UpdateEvent(Channel *channel)
    {
        bool ret = HasChannel(channel);
        if (ret == false) // 没有就添加
        {   
            m_channels[channel->Fd()] = channel;
            Update(channel, EPOLL_CTL_ADD);
        }
        else // 存在就修改
        {
            Update(channel, EPOLL_CTL_MOD);
        }
    }
    // 移除监控
    // void RemoveEvent(Channel *channel)
    // {
    //     auto it = m_channels.find(channel->Fd());
    //     if (it != m_channels.end())
    //     {
    //         m_channels.erase(it);
    //         Update(channel, EPOLL_CTL_DEL);
    //     }
    // }
    void RemoveEvent(Channel *channel)
    {
        auto it = m_channels.find(channel->Fd());
        if (it != m_channels.end())
        {
            /* 
                这两个删除有顺序要求，必须是先进行m_channels的删除
                因为如果先删除epoll中的节点的话，可能epoll不在管理该文件描述符，
                导致无法正确的查找Channel对象 
            */
            m_channels.erase(it);
        }
        Update(channel, EPOLL_CTL_DEL);
    }

    // 开始监控，返回活跃连接
    void Poll(std::vector<Channel *> *active)
    {
        int nfds = epoll_wait(m_epfd, m_evs, MAX_POLLEREVENTS, -1);
        if (nfds < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            ERR_LOG("EPOLL WAIT FAILED: %s", strerror(errno));
            abort();
        }
        for (int i = 0; i < nfds; i++)
        {
            auto it = m_channels.find(m_evs[i].data.fd);
            assert(it != m_channels.end());
            it->second->SetREvents(m_evs[i].events);
            active->push_back(it->second);
        }
    }
};


/******************************** 
            定时任务类
********************************/
class EventLoop;
using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>; 
class TimerTask 
{
private:
    uint64_t m_id;
    uint32_t m_timerefresh;
    TaskFunc m_task;
    ReleaseFunc m_release;
    bool m_canceled; //  定时任务是否被取消 --- 不能再时间轮中取消，因为这样的话相当于是提前执行了
public:
    TimerTask(uint64_t id, uint32_t timerefresh, TaskFunc task)
        :m_id(id)
        ,m_timerefresh(timerefresh)
        ,m_task(task)
        ,m_canceled(false)
    {}
    void Cancel()
    {
        m_canceled = true;
    }
    void SetReleaseFuc(const ReleaseFunc& release)
    {
        m_release = release;
    }
    uint32_t Delay()
    {
        return m_timerefresh;
    }
    ~TimerTask()
    {
        if (m_canceled == false)
            m_task();
        m_release();
    }
};


/******************************** 
            时间轮类
********************************/
class TimerWheel
{
private:
    using PtrTask = std::shared_ptr<TimerTask>;
    using WeakTask = std::weak_ptr<TimerTask>;
    int m_tick;
    int m_capacity;
    std::vector<std::vector<PtrTask>> m_wheel;
    std::unordered_map<uint64_t, WeakTask> m_timers;

    int m_timerfd; // 定时器描述符
    EventLoop *m_loop; // 监控timerfd
    std::unique_ptr<Channel> m_timer_channel; 
private:
    void RemoveTimer(uint32_t id)
    {
        auto it = m_timers.find(id);
        if (it != m_timers.end())
        {
            m_timers.erase(it);
        }
    }
    static int CreateTimerfd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            ERR_LOG("Create Timerfd Failed");
            abort();
        }
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0;
        timerfd_settime(timerfd, 0, &itime, nullptr);
        return timerfd;
    }
    int ReadTimerfd()
    {
        // 有可能因为其他描述符时间花费时间长，然后在处理定时器描述符的时候，有可能已经超时了很多次
        // read读取到的数据times就是上一次read之后超时的次数
        uint64_t times;
        int ret = read(m_timerfd, &times, 8);
        if (ret < 0)
        {
            ERR_LOG("Read Timerfd Failed!");
            abort();
        }
        return times;
    }
    void RunTimerTask()
    {

        m_tick = (m_tick + 1) % m_capacity;
        m_wheel[m_tick].clear();
    }
    void OnTime()
    {
        int times = ReadTimerfd();
        for(int i=0;i<times;i++)
            RunTimerTask();
    }
    void TimerAddInLoop(uint64_t id, uint32_t delaytime, TaskFunc task)
    {
        // 构建延时任务类
        PtrTask sp(new TimerTask(id, delaytime, task)); // 构建sharedptr，插入表盘中
        sp->SetReleaseFuc(std::bind(&TimerWheel::RemoveTimer, this, id));
        WeakTask wp(sp); // 构建weakptr，插入所有的延时队列中
        // 添加任务
        int inserti = (delaytime + m_tick) % m_capacity;
        m_wheel[inserti].push_back(sp);
        m_timers[id] = wp;
    }
    void TimerRefreshInLoop(uint64_t id)
    {
        auto it = m_timers.find(id);
        if (it == m_timers.end())
            return;
        PtrTask sp = it->second.lock();
        int inserti = (m_tick + sp->Delay()) % m_capacity;
        m_wheel[inserti].push_back(sp);
    }
    void TimerCancelInLoop(uint64_t id)
    {
        auto it = m_timers.find(id);
        if (it == m_timers.end())
            return;
        PtrTask sp = it->second.lock();
        if (sp) 
            sp->Cancel();
    }
public:
    TimerWheel(EventLoop *loop)
        :m_tick(0)
        ,m_capacity(60)
        ,m_wheel(m_capacity)
        ,m_timerfd(CreateTimerfd())
        ,m_loop(loop)
        ,m_timer_channel(new Channel(m_loop, m_timerfd))
    {
        m_timer_channel->SetReadCallBack(std::bind(&TimerWheel::OnTime, this));
        m_timer_channel->EnableRead();
    }
    /* 考虑线程安全: 如果不想加锁，就把对定期的所有操作都放到一个线程中执行 */
    void TimerAdd(uint64_t id, uint32_t delaytime, TaskFunc task);
    void TimerRefresh(uint64_t id);
    void TimerCancel(uint64_t id);
    /* 存在线程安全问题 --- 只能在组件内线程中使用 */
    bool HasTimer(uint64_t id)
    {
        auto it = m_timers.find(id);
        if (it == m_timers.end())
            return false;
        return true;
    }
    void TimerRun()
    {
        m_tick = (m_tick + 1) % m_capacity;
        m_wheel[m_tick].clear();
    }
    ~TimerWheel()
    {}
};


/******************************** 
    Eventloop: One Thread one Loop
********************************/
class EventLoop
{
private:
    std::thread::id m_thread_id; // 线程ID

    int m_event_fd; // eventfd唤醒所有的事件监控有可能导致阻塞
    std::unique_ptr<Channel> m_event_channel;
    Poller m_poller; // 对所有的描述符进行监控
   
    using Functor = std::function<void()>;
    std::vector<Functor> m_tasks; // 任务池
   
    std::mutex m_mutex; // 实现任务池操作的线程安全
    TimerWheel m_timer_wheel; // 定时器模块
public:
    void RunAllTask()
    {
        std::vector<Functor> tmp;
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_tasks.swap(tmp);
        }
        for (auto& t : tmp)
        {
            t();
        }
    }
    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (efd < 0)
        {
            ERR_LOG("create eventfd failed");
            abort();
        }
        return efd;
    }
    void ReadEvent()
    {
        uint64_t res = 0;
        int n = read(m_event_fd, &res, sizeof(res));
        if (n < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
                return;
            ERR_LOG("Read Event Failed!");
            abort();
        }
        return;
    }
    void WakeUp()
    {
        uint64_t res = 1;
        int n = write(m_event_fd, &res, sizeof(res));
        if (n < 0)
        {
            if (errno == EINTR || errno == EAGAIN)
                return;
            ERR_LOG("Write Event Failed!");
            abort();
        }
        return;
    }
public:
    EventLoop() 
        :m_thread_id(std::this_thread::get_id())
        ,m_event_fd(CreateEventFd()) // efd
        ,m_event_channel(new Channel(this, m_event_fd))
        ,m_timer_wheel(this)
    {
        m_event_channel->SetReadCallBack(std::bind(&EventLoop::ReadEvent, this));
        m_event_channel->EnableRead();
    }
    ~EventLoop() {}
    void Start() // 三步走，事件监控 ===> 就绪事件处理 ===> 执行任务
    {
        while (1)
        {
            std::vector<Channel *> actives;
            m_poller.Poll(&actives);
            for (auto& a : actives)
            {
                a->HandleEvent();
            }
            RunAllTask();
        }
    }
    void RunInLoop(const Functor& cb) // 判断要执行的任务是否处于当前线程，是则执行，否则压入任务池
    {
        if (IsInLoop())
        {
            // 必须return返回，要不就也执行下面的函数，导致多添加了一次cb
            return cb();
        }
        return QueueInLoop(cb);
    }
    void QueueInLoop(const Functor& cb) // 将操作压入任务池
    {
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_tasks.push_back(cb);
        }
        WakeUp();
    }
    bool AssertInLoop()
    {
       assert(m_thread_id == std::this_thread::get_id());
    }
    bool IsInLoop() // 判断操作是否在当前线程
    {
        return m_thread_id == std::this_thread::get_id();
    }
    void UpdateEvent(Channel *channel) { return m_poller.UpdateEvent(channel); } // 添加/修改描述符的事件监控
    void RemoveEvent(Channel *channel) { return m_poller.RemoveEvent(channel); } // 删除描述符的监控
    void TimerAdd(uint64_t id, uint32_t delay, TaskFunc task) { return m_timer_wheel.TimerAdd(id, delay, task); }
    void TimerRefresh(uint64_t id) { return m_timer_wheel.TimerRefresh(id); } 
    bool HasTimer(uint64_t id) { return m_timer_wheel.HasTimer(id); }
    void TimerCancel(uint64_t id) { return m_timer_wheel.TimerCancel(id); }
};


/******************************** 
            LoopThread
********************************/
class LoopThread
{
private:
    std::mutex m_mutex;
    std::condition_variable m_cond;
    EventLoop *m_loop;          // EventLoop指针变量，这个对象在线程内初始化
    std::thread m_thread;       // EventLoop对应的线程

private:
    /* 实例化EventLoop对象，唤醒cond上有可能阻塞的线程，并且开始运行EventLoop模块功能 */
    void ThreadEntry()
    {
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_loop = &loop;
            m_cond.notify_all();
        }
        loop.Start();
    }
public:
    /* 创建线程，设定线程入口函数 */
    LoopThread() 
        :m_loop(nullptr)
        ,m_thread(std::bind(&LoopThread::ThreadEntry, this))
    {}
    ~LoopThread() {}
    /* 返回当前线程关联的EventLoop指针对象 */
    EventLoop *GetLoop()
    {
        EventLoop *loop = nullptr;
        {
            std::unique_lock<std::mutex> lock(m_mutex);
            m_cond.wait(lock, [&](){ return m_loop != nullptr; });
            loop = m_loop;
        }
        return loop;
    }
};


/******************************** 
        LoopThreadPool
********************************/
class LoopThreadPool
{
private:
    int m_thread_count;
    int m_next_idx;
    EventLoop *m_baseloop;
    std::vector<LoopThread *> m_threads;
    std::vector<EventLoop *> m_loops;
public:
    LoopThreadPool(EventLoop *loop)
        :m_thread_count(0)
        ,m_next_idx(0)
        ,m_baseloop(loop)
    {}
    ~LoopThreadPool() {}
    void SetThreadCount(int count)
    {
        m_thread_count = count;
    }
    void Create()
    {
        if (m_thread_count > 0)
        {
            m_threads.resize(m_thread_count);
            m_loops.resize(m_thread_count);
            for (int i = 0; i < m_thread_count; i++)
            {
                m_threads[i] = new LoopThread();
                m_loops[i] = m_threads[i]->GetLoop();
            }
        }
        return;
    }
    EventLoop *NextLoop()
    {
        if (m_thread_count == 0)
            return m_baseloop;
        m_next_idx = (m_next_idx + 1) % m_thread_count;
        return m_loops[m_next_idx];
    }
};



/******************************** 
            Any
********************************/
class Any
{
private:
    class Holder
    {
    public:
        Holder() {}
        virtual const std::type_info& Type() = 0;
        virtual Holder *Clone() = 0;
        ~Holder() {}
    };
    template<class T>
    class PlaceHolder : public Holder
    {
    public:
        PlaceHolder(T val) :m_val(val) {}
        const std::type_info& Type()
        {
            return typeid(T);
        }
        // ????
        Holder* Clone()
        {
            return new PlaceHolder(m_val);
        }
        ~PlaceHolder() {}
    public:
        T m_val;
    };
    Holder *m_content;
public:
    Any() :m_content(nullptr) {}
    template<class T>
    Any(const T& val) :m_content(new PlaceHolder<T>(val)) {}
    Any(const Any& other) :m_content(other.m_content->Clone()) {}
    ~Any() { delete m_content; }
    Any& Swap(Any& other)
    {
        std::swap(m_content, other.m_content);
        return *this;
    }
    template<class T>
    // 获取内容
    T* Get() 
    {
        if (typeid(T) != m_content->Type())
            return nullptr;
        return &dynamic_cast<PlaceHolder<T>*>(m_content)->m_val;
    }
    template<class T>
    Any& operator=(const T& val) 
    {
        Any(val).Swap(*this);
        return *this;
    }
    Any& operator=(const Any& other) 
    {
        m_content = other.m_content->Clone();
        return *this;
    }
};


/******************************** 
        Connection
********************************/
class Connection;
typedef enum 
{
    DISCONNECTED, // 连接关闭状态
    CONNECTING, // 连接创建成功状态 --- 待处理状态
    CONNECTED, // 连接建立完成状态 --- 连接建立完成，各种设置完成，可以通信状态
    DISCONNECTING // 待关闭状态
}ConnStatus;
using PtrConnection = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    uint64_t m_conn_id; // 连接的唯一ID，便于连接的管理和查找
    // uint64_t m_timerfd_id; // 定时器ID，为了简便和连接ID保持一致
    int m_sockfd; // 连接关联文件的描述符
    bool m_enable_inactive_release; // 是否启动非活跃连接释放
    ConnStatus m_status; // 连接状态
    Socket m_socket; // 套接字管理
    EventLoop *m_loop; // 连接所关联的EventLoop
    Channel m_channel; // 连接的事件管理
    Buffer m_in_buffer; // 输入缓冲区 --- 存放socket中读取的数据
    Buffer m_out_buffer; // 输出缓冲区 --- 存放要发送给对端的消息
    Any m_context; // 请求接收处理上下文

    /*
        这四个回调函数，是让服务器模块来设置的，（其实服务器模块的处理回调也是组件使用者设置的）
        ----> 这几个回调函数都是组件使用者使用的
    */
    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    ConnectedCallback m_connected_callback;
    MessageCallback m_message_callback;
    ClosedCallback m_closed_callback;
    AnyEventCallback m_event_callback;
    /*
        组件内的连接关闭回调---组件内设置的，因为服务器组件会把所有的连接管理起来，一旦某个连接关闭
        就该从管理的地方移除掉自己的信息
    */
    ClosedCallback m_server_closed_callback;
private:
    /* 五个channel的事件回调 */
    void HandleRead() // 描述符可读事件触发后调用的函数，接收socket数据放到接收缓冲区，然后调用_message_callback
    {
        // 1、从socket中接收数据
        char buf[65536] = {0}; // 4KB
        int ret = m_socket.NonBlockRecv(buf, 65535);
        if (ret < 0)
        {
            // 接收失败, 出错了，不能直接关闭连接。
            // 需要先检查有没有数据待发送，如果没有数据，直接关闭连接，有数据就发送数据。
            return ShutdownInLoop();
        }
        else if (ret == 0)
        {
            // 没有数据
            return;
        }
        // 2、将数据写入到输入缓冲区
        // 写入之后顺便将写偏移进行移动
        m_in_buffer.WriteAndPush(buf, ret);
        // 3、输入缓冲区有数据才调用message回调函数
        if (m_in_buffer.ReadAbleSize() > 0)
        {
            // shared_from_this --- 从当前对象自身获取自身的shared_ptr管理对象 
            return m_message_callback(shared_from_this(), &m_in_buffer);  
        }
    }
    void HandleWrite() //描述符触发可写事件，将发送缓冲区的数据进行发送 
    {
        // 进行数据的发送
        ssize_t ret = m_socket.NonBlockSend(m_out_buffer.ReaderPosition(), m_out_buffer.ReadAbleSize());
        // 如果发送数据错误，释放连接连接
        if (ret < 0)
        {
            // 如果输入缓冲区有数据的话，需要进行回调
            if (m_in_buffer.ReadAbleSize() > 0)
            {
                DBG_LOG("m_message_callback");
                m_message_callback(shared_from_this(), &m_in_buffer);
            }
            return Release();
        }
        // 因为我们直接调用的是原始的接口，没有进行读偏移的移动，所以我们需要手动进行读偏移的移动
        // 为什么移动的是读偏移，因为我们是把输出缓冲区的数据发送出去
        m_out_buffer.MoveReadOffset(ret);

        // 我们需要判断可读数据的大小，如果可读数据为0, 且为待释放连接, 则释放连接
        if (m_out_buffer.ReadAbleSize() == 0)
        {
            // 没有数据发送，关闭写事件关心
            m_channel.DisableWrite();
            // 如果为待释放连接，则释放连接
            if (m_status == DISCONNECTING)
            {
                return Release();
            }
        }
        return;
    }
    void HandleClose() // 描述符触发挂断事件
    {
        // 如果有数据待处理就处理一下，完毕释放连接
        if (m_in_buffer.ReadAbleSize() > 0)
        {
            DBG_LOG("m_message_callback");
            m_message_callback(shared_from_this(),&m_in_buffer);
        }
        return Release();
    }
    void HandleError() // 描述符触发异常事件
    {
        return HandleClose();
    }
    void HandleEvent() // 描述符触发任意事件
    {
        // 刷新活跃度
        if (m_enable_inactive_release == true)
        {
            m_loop->TimerRefresh(m_conn_id);
        }
        // 任意事件回调
        if (m_event_callback)
        {
            m_event_callback(shared_from_this());
        }
    }

    void EstablishedInLoop() // 连接获取之后，所处的状态下要进行各种设置（给channel设置事件回调，启动读监控）
    {
        // 1、修改状态
        assert(m_status == CONNECTING);
        m_status = CONNECTED;
        // 2、启动读事件监控
        m_channel.EnableRead();
        // 3、调用回调函数
        if (m_connected_callback) 
            m_connected_callback(shared_from_this());
    }
    void ReleaseInLoop() // 实际的释放接口
    {
        // 1、修改连接状态
        m_status = DISCONNECTED;
        // 2、移除事件监控
        m_channel.Remove();
        // 3、关闭描述符
        m_socket.Close();
        // 4、取消定时任务
        if (m_loop->HasTimer(m_conn_id)) CancelInactiveReleaseInLoop();
        // 5、关闭回调函数
        // 先调用用户设置的关闭回调
        if (m_closed_callback) m_closed_callback(shared_from_this());
        // 再调用系统调用的关闭回调
        if (m_server_closed_callback) m_server_closed_callback(shared_from_this());
    }
    void Release()
    {
        m_loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));
    }
    void SendInLoop(Buffer& buf) // 发送数据，将数据放到缓冲区中，启动写事件监控
    {
        // 需要判断状态，如果是关闭状态就直接返回
        if (m_status == DISCONNECTED)
            return;
        m_out_buffer.WriteBufferAndPush(buf);
        if (!m_channel.WriteAble())
            m_channel.EnableWrite();
    }
    void ShutdownInLoop() // 提供给组件使用者的关闭连接接口---并不实际关闭，需要判断有无数据待处理
    {
        // 1、先修改状态
        m_status = DISCONNECTING;
        // 2、如果输入缓冲区有数据调用输入回调
        if (m_in_buffer.ReadAbleSize() > 0)
        {
            DBG_LOG("%ld", m_in_buffer.ReadAbleSize());
            if (m_message_callback) 
            {
                m_message_callback(shared_from_this(), &m_in_buffer);
               DBG_LOG("m_message_callback");
            }
        }
        // 3、如果输出缓冲区有数据，启动对写事件监控
        if (m_out_buffer.ReadAbleSize() > 0)
        {
            if (!m_channel.WriteAble())
                m_channel.EnableWrite();
        }
        if (m_out_buffer.ReadAbleSize() == 0)
            Release();
    }
    void EnableInactiveReleaseInLoop(int sec /*seconds*/) // 启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
    {
        // 1、更新标识位
        m_enable_inactive_release = true;
        // 2、如果存在定时任务，就刷新
        if (m_loop->HasTimer(m_conn_id))
            return m_loop->TimerRefresh(m_conn_id);
        // 3、如果不存在定时任务，就增加
        return m_loop->TimerAdd(m_conn_id, sec, std::bind(&Connection::Release, this));
    }
    void CancelInactiveReleaseInLoop() // 取消非活跃销毁
    {
        // 1、更新标志位
        m_enable_inactive_release = false;
        // 2、如果存在就取消
        if (m_loop->HasTimer(m_conn_id))
            m_loop->TimerCancel(m_conn_id);
    }
    void UpgradeInLoop(const Any& context, 
                const ConnectedCallback& conn, 
                const MessageCallback& msg, 
                const ClosedCallback& closed, 
                const AnyEventCallback& event) // 切换协议 --- 重置上下文以及阶段性处理函数
    {
        m_context = context;
        m_message_callback = msg;
        m_connected_callback = conn;
        m_closed_callback = closed;
        m_event_callback = event;
    }
public:
    Connection(EventLoop *loop, uint64_t conn_id, int sockfd) 
        :m_conn_id(conn_id)
        ,m_sockfd(sockfd)
        ,m_enable_inactive_release(false)
        ,m_status(CONNECTING)
        ,m_socket(m_sockfd)
        ,m_loop(loop)
        ,m_channel(loop, sockfd)
    {
        m_channel.SetCloseCallBack(std::bind(&Connection::HandleClose, this));
        m_channel.SetEventCallBack(std::bind(&Connection::HandleEvent, this));
        m_channel.SetErrorCallBack(std::bind(&Connection::HandleError, this));
        m_channel.SetReadCallBack(std::bind(&Connection::HandleRead, this));
        m_channel.SetWriteCallBack(std::bind(&Connection::HandleWrite, this));
    }
    ~Connection() { DBG_LOG("Release Connection: %p", this); }
    int Fd()
    {
        return m_sockfd;
    }
    int Id()
    {
        return m_conn_id;
    }
    bool Connected() // 是否处于CONNECTED状态 
    {
        return m_status == CONNECTED;
    }
    void SetContext(const Any& context) // 设置上下文数据 --- 连接建立完成后调用
    {
        m_context = context;
    }
    Any *GetContext() // 获取上下文 --- 返回的是指针
    {
        return &m_context;
    }
    void SetConnectedCallback(const ConnectedCallback& cb)
    {
        m_connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback& cb)
    {
        m_message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback& cb)
    {
        m_closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback& cb)
    {
        m_event_callback = cb;
    }
    void SetSrvClosedCallback(const ClosedCallback& cb)
    {
        m_server_closed_callback = cb;
    }
    void Established() // 连接建立后，进行channel回调设置，启动读监控，调用m_connected_callback
    {
        m_loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }
    void Send(const char *data, size_t len) // 发送数据，将数据放到缓冲区中，启动写事件监控
    {
        Buffer buf;

        buf.WriteAndPush(data, len);
        return m_loop->RunInLoop(std::bind(&Connection::SendInLoop, this, std::move(buf)));
    }
    void Shutdown() // 提供给组件使用者的关闭连接接口---并不实际关闭，需要判断有无数据待处理
    {
        return m_loop->RunInLoop(std::bind(&Connection::ShutdownInLoop, this));
    }
    void EnableInactiveRelease(int sec /*seconds*/) // 启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
    {
        return m_loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop, this, sec));
    }
    void CancelInactiveRelease() // 取消非活跃销毁
    {
        return m_loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    void Upgrade(const Any& context, 
                const ConnectedCallback& conn, 
                const MessageCallback& msg, 
                const ClosedCallback& closed, 
                const AnyEventCallback& event) // 切换协议 --- 重置上下文以及阶段性处理函数
    {
        m_loop->AssertInLoop();
        return m_loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, msg, closed, event));
    }
};


/******************************** 
        Acceptor
********************************/
class Acceptor
{
private:
    Socket m_socket; // 套接字管理
    EventLoop *m_loop; // 事件监控
    Channel m_channel; // 事件管理

    using AcceptCallback = std::function<void(int)>;
    AcceptCallback m_accept_callback;
private:
    /* 调用m_accept_callback函数进行新连接管理 */
    void HandleRead()
    {
        int newfd = m_socket.Accept();
        if (newfd < 0)
            return;
        if (m_accept_callback) 
            m_accept_callback(newfd);
    }
    int CreateServer(int port)
    {
        bool ret = m_socket.CreateServer(port);
        assert(ret == true);   

        return m_socket.Fd();
    }
public:
    /* 不能在创建的时候就启动事件监控，可能当事件来的时候，回调函数还没有设置，导致没有调用回调函数，资源泄露 */
    Acceptor(EventLoop *loop, int port) 
        :m_socket(CreateServer(port))
        ,m_loop(loop)
        ,m_channel(loop, m_socket.Fd())
    {
        m_channel.SetReadCallBack(std::bind(&Acceptor::HandleRead, this));
    }
    ~Acceptor() {}

    void SetAcceptCallback(const AcceptCallback& cb)
    {
        m_accept_callback = cb;
    }
    void Listen()
    {
        m_channel.EnableRead();
    }
};


/******************************** 
        TcpServer
********************************/
class TcpServer
{
private:
    uint64_t m_conn_id; // 自动增长的连接ID
    int m_port; // 端口
    int m_timeout; // 非活跃销毁任务的时间
    bool m_enable_inactive_release; // 是否启动了非活跃销毁任务
    EventLoop m_baseloop; // 主线程的EventLoop对象，负责监听事件的处理
    Acceptor m_acceptor; // 监听套接字的管理对象
    LoopThreadPool m_pool; // 从属线程池
    std::unordered_map<uint64_t, PtrConnection> m_conns; // 保存所有连接

    using ConnectedCallback = std::function<void(const PtrConnection&)>;
    using MessageCallback = std::function<void(const PtrConnection&, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection&)>;
    using AnyEventCallback = std::function<void(const PtrConnection&)>;
    ConnectedCallback m_connected_callback;
    MessageCallback m_message_callback;
    ClosedCallback m_closed_callback;
    AnyEventCallback m_event_callback;
private:
    using Functor = std::function<void()>;
    void RunAfterInLoop(const Functor& task, int delay)
    {
        m_conn_id++;
        m_baseloop.TimerAdd(m_conn_id, delay, task);
    }
    void RemoveConnectionInLoop(const PtrConnection& conn) // 从管理Connection的m_conns中移除信息
    {
        int id = conn->Id();
        auto it = m_conns.find(id);

        if (it != m_conns.end())
        {
            m_conns.erase(it);
        }
    }
    void RemoveConnection(const PtrConnection& conn) // 从管理Connection的m_conns中移除信息
    {
        DBG_LOG("Release Connection: %p",conn);
        m_baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this, conn));
    }
    void NewConnection(int fd) // 为新连接创建一个Connection对象
    {
        m_conn_id++;
        PtrConnection conn(new Connection(m_pool.NextLoop(), m_conn_id, fd));
        conn->SetMessageCallback(m_message_callback);
        conn->SetConnectedCallback(m_connected_callback);
        conn->SetClosedCallback(m_closed_callback);
        conn->SetAnyEventCallback(m_event_callback);
        conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection, this, conn));

        if (m_enable_inactive_release)
            conn->EnableInactiveRelease(m_timeout);
        conn->Established();
        m_conns.insert(std::make_pair(m_conn_id, conn));
    }
public:
    TcpServer(int port) 
        :m_conn_id(0)
        ,m_port(port)
        ,m_enable_inactive_release(false)
        ,m_acceptor(&m_baseloop, m_port)
        ,m_pool(&m_baseloop)
    {
        m_acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        m_acceptor.Listen();
    }
    ~TcpServer() {}
    void SetThreadCount(int count)
    {
        m_pool.SetThreadCount(count);
    }
    void Start()
    {
        m_pool.Create();
        m_baseloop.Start();
    }
    void SetConnectedCallback(const ConnectedCallback& cb)
    {
        m_connected_callback = cb;
    }
    void SetMessageCallback(const MessageCallback& cb)
    {
        m_message_callback = cb;
    }
    void SetClosedCallback(const ClosedCallback& cb)
    {
        m_closed_callback = cb;
    }
    void SetAnyEventCallback(const AnyEventCallback& cb)
    {
        m_event_callback = cb;
    }
    void EnableInactiveRelease(int timeout)
    {
        m_timeout = timeout;
        m_enable_inactive_release = true;
    }
    void RunAfter(const Functor& task, int delay) // 添加一个定时任务
    {
        m_baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop, this, task, delay));
    }
};

/* Channel */
void Channel::Remove() { m_loop->RemoveEvent(this); }
void Channel::Update() { m_loop->UpdateEvent(this); }
/* TimerWheel */
/* 考虑线程安全: 如果不想加锁，就把对定期的所有操作都放到一个线程中执行 */
void TimerWheel::TimerAdd(uint64_t id, uint32_t delaytime, TaskFunc task)
{
    return m_loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop, this, id, delaytime, task));
}
void TimerWheel::TimerRefresh(uint64_t id)
{
    return m_loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop, this, id));
}
void TimerWheel::TimerCancel(uint64_t id)
{
    return m_loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop, this, id));
}


class NetWork
{
public:
    NetWork() 
    {
        DBG_LOG("SIGPIPE INIT");
        signal(SIGPIPE, SIG_IGN);
    }
};
static NetWork nw;